package com.ruoyi.web.controller.kafka;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.utils.ToolUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

/**
 * @Description:
 * @Author: LJ
 * @Date: 2022/9/23 11:16
 * @version:
 */

@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    //自定义topic
    public static final String TOPIC_TEST = "topic_test";

    public void send(Object obj) {
        //发送消息
        sendMessage(TOPIC_TEST, "1", obj);
    }

    /**
     * 发送消息
     *
     * @param topic
     * @param message
     */
    public void sendMessage(String topic, String key, Object message) {
        this.sendMessage(topic, key, parseMessage(message));
    }

    private String parseMessage(Object message) {
        if (message == null) return null;
        if (message instanceof String) return message.toString();
        if (message instanceof Integer) return message.toString();
        if (message instanceof Long) return message.toString();
        return JSONObject.toJSONString(message);
    }

    /**
     * 发送消息
     *
     * @param topic   消息主题
     * @param key     分区唯一
     * @param message 消息美容
     */
    private void sendMessage(String topic, String key, String message) {
        if (ToolUtil.isEmpty(topic)) {
            log.error("消息主题不能为空, topic={}, key={}, message={}", topic, key, ToolUtil.abbr(message, 50));
        } else {
            log.debug("发送消息Kafka sent->> topic={}, key={}, message={}", topic, key, ToolUtil.abbr(message, 50));
            kafkaTemplate.send(topic, key, message);
        }
    }
}
